package sun.cn.common.holder;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * 并行任务处理器
 * 
 * @author sunlingao@zbj.com
 * @title
 * @date 2017年8月18日
 * @version
 */
public final class ConcurrentHolder {

	private static final Logger LOG = LoggerFactory.getLogger(ConcurrentHolder.class);

	private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(
			Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(), 1000,
			TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024),
			new ThreadFactoryBuilder().setNameFormat("concurrent-%s").build(), new ThreadPoolExecutor.AbortPolicy());

	private ConcurrentHolder() {}

	/**
	 * 并行处理器
	 * 
	 * @author sunlingao@zbj.com
	 * @date 2017年8月18日
	 * @version
	 * @param tasks
	 *            任务队列
	 * @param ret
	 *            返回类型
	 * @return
	 */
	public static <T> List<T> ConcurrentTask(List<ConcurrentTask> tasks, Class<T> ret) {
		return ConcurrentTask(tasks, 0, null, ret);
	}

	/**
	 * 并行处理器 不返回结果
	 * 
	 * @author sunlingao@zbj.com
	 * @date 2017年8月18日
	 * @version
	 * @param tasks
	 *            任务队列
	 * @return
	 */
	public static void ConcurrentTask(List<ConcurrentTask> tasks) {
		ConcurrentTask(tasks, 0, null, null);
	}

	/**
	 * 并行处理器 不返回结果
	 * 
	 * @author sunlingao@zbj.com
	 * @date 2017年8月18日
	 * @version
	 * @param tasks
	 *            任务队列
	 * @param time
	 *            最长处理时间
	 * @param unit
	 *            时间单位
	 * @return
	 */
	public static void ConcurrentTask(List<ConcurrentTask> tasks, long time, TimeUnit unit) {
		ConcurrentTask(tasks, time, unit, null);
	}

	/**
	 * 并行处理器
	 * 
	 * @author sunlingao@zbj.com
	 * @date 2017年8月18日
	 * @version
	 * @param tasks
	 *            任务队列
	 * @param time
	 *            最长处理时间
	 * @param unit
	 *            时间单位
	 * @param ret
	 *            返回类型
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static <T> List<T> ConcurrentTask(List<ConcurrentTask> tasks, long time, TimeUnit unit, Class<T> ret) {
		final CountDownLatch latch = new CountDownLatch(tasks.size());
		List<T> result = (ret == null ? null : Lists.newArrayList());
		for (final ConcurrentTask task : tasks) {
			EXECUTOR.execute(() -> {
				try {
					T obj = (T) task.task();
					if (ret != null)
						result.add(obj);
				} catch (Exception e) {
					LOG.error("ConcurrentTask error", e);
					try {
						if (ret != null)
							result.add(ret.newInstance());
					} catch (Exception e1) {
						LOG.error("ConcurrentTask error getInstance is {} e is {}", ret.getName(), e1);
					}
				} finally {
					latch.countDown();
				}
			});
		}
		try {
			if (time > 0) {
				latch.await(time, unit);
			} else {
				latch.await();
			}
		} catch (Exception re) {
			LOG.error("ConcurrentTask error", re);
		}
		return result;
	}

	public interface ConcurrentTask {
		public Object task();

	}
}
